package com.jiesone.logger.distributecleaning.handler;
/*
 *@ClassName DataProcessingHandler
 *@Description          通过消息队列或者Redis数据库中获取日志源数据 进行数据处理并存储到数据库中
 *@Author tangsixiang@163.com
 *@Date 2022/1/17 9:53
 *@Version 1.0
 */


import com.alibaba.fastjson.JSON;
import com.jiesone.logger.distributecleaning.config.CommonRediskey;

import com.jiesone.logger.distributecleaning.util.ElasticSearchUtil;
import com.jiesone.logger.distributecleaning.util.JsonStringUtil;
import com.jiesone.logger.distributecleaning.util.SnowFlake;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;
import java.util.HashMap;
import java.util.Map;


/***
 *  从MQ获取目标数据源 再次Hanlder中数据分拣---->分流---->存储目标数据
 */
@Component
public class DataProcessingHandler {

    static  Logger logger=  LoggerFactory.getLogger(DataProcessingHandler.class);

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
     private ElasticSearchUtil searchUtil;

    @Autowired
    SnowFlake snowFlake;



    @Value("${logger-filebeat.discardKey}")


    private String discardKey;


    @Value(("${Logger.level}"))
    private  String loggerLevel;

    @Autowired
      CommonRediskey rediskey;




    public  void   pullApartStringMessage(String serviceCode,String sourceData){


        //1 获取消息归属sourceCode;  根据消息头可以至获取到目标数据
       boolean exsites=stringRedisTemplate.opsForHash().hasKey(CommonRediskey.accessSource,serviceCode);
        logger.info(serviceCode+"判断是否拥有是否需要处理数值："+exsites);
        if(exsites){
            //2 通过消息SourceCode 保存源数据
            saveDataSourceByMQ(serviceCode,sourceData);
            //3、根据消息链 数据格式类型 走具体数据清洗规则业务处理




        }else {
            logger.info("--------废弃数据无需处理--业务代码:"+serviceCode);
        }

    }





    public  void  saveDataSourceByMQ(String sourceCode,String  sourceData){

        //走数据默认存储
         String  key=CommonRediskey.savaTarget;
         logger.debug("------进入纯粹数据工作：key:"+key);
              if(stringRedisTemplate.opsForHash().hasKey(key,sourceCode+":1")){
                  try {
                      //  获取 存储log源数据信息
                      Object   iii =stringRedisTemplate.opsForHash().entries(key).get(sourceCode + ":1");
                     Map<String,Object> targetMap=  JSON.parseObject(iii.toString(),HashMap.class);
                       String  indexName=   String.valueOf(targetMap.get("targetIndex"));
                      Map<String,Object> targetData=      JSON.parseObject(sourceData,HashMap.class);
                        //移除掉 Agent传输过程中 自追加系统字段
                          for (String  dropKey:discardKey.split(",")){
                              targetData.remove(dropKey);
                       }
                      targetData.remove("@timestamp");targetData.remove("@metadata");

         //      logger.debug("-----源数据存储：--------"+targetData);
                        Long  snowId=snowFlake.nextId();
                      targetData.put("timestamp", System.currentTimeMillis());
                  //    targetData.put("id",snowId);
                      //判断索引存储是否按照规整调整     N 不区分  H 小时 D 天 M 月Y 年  格式创建索引

                String  level=JsonStringUtil.checkLoggerLevel(targetData.get("message").toString());
                          if(level!=null){
                              targetData.put("logLevel",level);
                          }
                      searchUtil.insertOrUpdateOne(JsonStringUtil.checkStrForValue(indexName,targetMap.get("targetData")),snowId,targetData);
                      targetData.put("sourceCode",sourceCode);

                    //  针对数据整合规则调用采用后置方式进行处理    采用

                      //默认日志规则保存数据库 便于浏览日志文件
                      this.resolveDataSourceCheck(sourceCode,"Loger:dataResolve",targetData);

                  //盘点是否有源日志内容告警规则设置
                      if(targetData.get("message").toString().length()>=450){
                          this.logAlarmNotification(1,level,indexName,sourceCode,snowId,targetData.get("message").toString().substring(0,450));
                      }else{
                          this.logAlarmNotification(1,level,indexName,sourceCode,snowId,targetData.get("message").toString());

                      }


                       }catch ( Exception e){
                    e.printStackTrace();
                       }

        }


    }
    /**
     * 分解清洗日志数据
     * @param channel  订阅通道类型
     * @param messageData
     */
    public  void  resolveDataSourceCheck(String sourceCode, String channel,  Map<String,Object>  messageData) {

         String key=CommonRediskey.savaTarget;
        // 走结构化存储解析--日志内容
        if (stringRedisTemplate.opsForHash().hasKey(key,sourceCode+":0")) {

            logger.debug("================清洗加工日志源文件内容=========");
            stringRedisTemplate.convertAndSend(channel, JSON.toJSONString(messageData));

        }

    }


    /**
     *
     * @param indexName 告警 索引名称
     * @param sourceCode  告警业务编码
     * @param snowId   告警查询字段（值） 默认使用 雪花算法生产ID
     * @return
     */
        public  boolean   logAlarmNotification(int  alarmType,String level,String indexName,String sourceCode,Long snowId,String mes){

              String alarmKey=sourceCode+":"+alarmType;
            String key=CommonRediskey.aleamRule;
            if(stringRedisTemplate.opsForHash().hasKey(key, alarmKey)&&
                    JsonStringUtil.checkLevel(stringRedisTemplate.opsForHash().get(key,alarmKey).toString(),level)){
                logger.info("=========根据设置的告警规则生产符合要求告警信息============:"+alarmKey);
           return  jdbcTemplate.update(
                   "INSERT INTO `loger_alarm_ruledesc` (`desc_id`,`alarm_type`,`alarm_level`,`index_name`,`source_code`,`query_value`,alarm_mes)VALUE(?,?,?,?,?,?,?)"
                                                             ,snowFlake.nextId(),alarmType,level,indexName,sourceCode,snowId,mes)>0;

            }
            return  false;
        }







}
